{
 "cells": [
  {
   "metadata": {},
   "cell_type": "markdown",
   "source": [
    "### 1. 环境配置\n",
    "安装依赖库"
   ],
   "id": "3ad74f7d3743c24b"
  },
  {
   "metadata": {
    "ExecuteTime": {
     "end_time": "2025-04-16T02:50:20.373008Z",
     "start_time": "2025-04-16T02:50:17.476257Z"
    }
   },
   "cell_type": "code",
   "source": "!pip install paho-mqtt==1.6.0",
   "id": "8f07f15c00a5ce2d",
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Looking in indexes: https://mirrors.aliyun.com/pypi/simple/\n",
      "Requirement already satisfied: paho-mqtt==1.6.0 in d:\\language environments\\python\\3.12.9\\lib\\site-packages (1.6.0)\n"
     ]
    }
   ],
   "execution_count": 11
  },
  {
   "metadata": {},
   "cell_type": "markdown",
   "source": [
    "### 2. 节点定义\n",
    "\n",
    "#### 2.1 必要配置\n",
    "1. 导入库"
   ],
   "id": "27a0f6c7b10a8552"
  },
  {
   "cell_type": "code",
   "id": "initial_id",
   "metadata": {
    "collapsed": true,
    "ExecuteTime": {
     "end_time": "2025-04-16T02:50:20.405897Z",
     "start_time": "2025-04-16T02:50:20.401801Z"
    }
   },
   "source": [
    "# python 3.12.9\n",
    "import random\n",
    "import time\n",
    "import json\n",
    "import threading as th\n",
    "from paho.mqtt import client as mqtt_client"
   ],
   "outputs": [],
   "execution_count": 12
  },
  {
   "metadata": {},
   "cell_type": "markdown",
   "source": "2. 使用免费的Broker带代理服务器，配置通信话题",
   "id": "198f38006930fac7"
  },
  {
   "metadata": {
    "ExecuteTime": {
     "end_time": "2025-04-16T02:50:20.421182Z",
     "start_time": "2025-04-16T02:50:20.416323Z"
    }
   },
   "cell_type": "code",
   "source": [
    "broker = 'broker.emqx.io'\n",
    "port = 1883\n",
    "topic = f\"/python/mqtt-{random.randint(10000,99999)}\"  # 自定义一个随机话题，避免和别人重复！否则会收到错误消息\n",
    "delay_status = dict()   # 每个接收节点的延迟情况统计，为-1则丢包，否则为具体延迟(ms)\n",
    "batch_size = 20         # 每次收发的数量\n",
    "msg_id = 0              # 发布节点每条消息的id，用来区分消息"
   ],
   "id": "63577e79884b19b2",
   "outputs": [],
   "execution_count": 13
  },
  {
   "metadata": {},
   "cell_type": "markdown",
   "source": "3. 定义连接MQTT服务器的函数",
   "id": "a1a7d2159a34958e"
  },
  {
   "metadata": {
    "ExecuteTime": {
     "end_time": "2025-04-16T02:50:20.438161Z",
     "start_time": "2025-04-16T02:50:20.431743Z"
    }
   },
   "cell_type": "code",
   "source": [
    "# 连接到MQTT Broker\n",
    "def connect_mqtt(client_id):\n",
    "    def on_connect(client, userdata, flags, rc):  # 发布/订阅端创建后连接broker会调用一次该（回调）函数\n",
    "        if rc == 0:\n",
    "            print(f\"{client_id} connected to MQTT Broker!\")\n",
    "        else:\n",
    "            print(f\"Failed to connect for {client_id}, return code {rc}\")\n",
    "\n",
    "    client = mqtt_client.Client(client_id)\n",
    "    client.on_connect = on_connect\n",
    "    client.connect(broker, port)\n",
    "    return client"
   ],
   "id": "977c98dac5797f70",
   "outputs": [],
   "execution_count": 14
  },
  {
   "metadata": {},
   "cell_type": "markdown",
   "source": "### 2.2 消息发布节点",
   "id": "491f205680f9797b"
  },
  {
   "metadata": {
    "ExecuteTime": {
     "end_time": "2025-04-16T02:50:20.457510Z",
     "start_time": "2025-04-16T02:50:20.450073Z"
    }
   },
   "cell_type": "code",
   "source": [
    "def publisher_node():\n",
    "    global msg_id\n",
    "    publisher_id = f'Publisher-{random.randint(0, 99):0>2}'\t# 发布设备的设定唯一设备号\n",
    "    client = connect_mqtt(publisher_id)\n",
    "    client.loop_start()  # 启动后台线程来处理网络循环\n",
    "\n",
    "    for i in range(batch_size):\n",
    "        time.sleep(0.1)\n",
    "        # 将消息数据转化为 JSON 格式\n",
    "        message_json = json.dumps({\n",
    "            'msg': msg_id,\n",
    "            'time': time.time(),\n",
    "        })\n",
    "\n",
    "        # 向指定topic发送信息\n",
    "        result = client.publish(topic, message_json)\n",
    "\n",
    "        # 解析响应内容\n",
    "        status = result[0]\n",
    "        if status == 0:\t# 发送成功\n",
    "            print(f\"{publisher_id}\\t发送: `{msg_id}` to topic `{topic}`\")\n",
    "        else:\t# 发送失败\n",
    "            print(f\"Failed to send message `{msg_id}`\")\n",
    "        msg_id += 1"
   ],
   "id": "92939c27b75f902b",
   "outputs": [],
   "execution_count": 15
  },
  {
   "metadata": {},
   "cell_type": "markdown",
   "source": [
    "### 2.3 消息订阅节点\n",
    "#### 2.3.1 延迟模拟"
   ],
   "id": "9d8a9d0b064b617b"
  },
  {
   "metadata": {
    "ExecuteTime": {
     "end_time": "2025-04-16T02:50:20.476231Z",
     "start_time": "2025-04-16T02:50:20.470005Z"
    }
   },
   "cell_type": "code",
   "source": [
    "def delay(client_id, received_msg):\n",
    "    global delay_status\n",
    "    # 每个设备的最大延迟为其id的后两位数字[0,99]与25取均值后再平方，取值范围为[12,62]之间整数的平方，单位ms\n",
    "    time.sleep(random.randint(1, int(25+int(client_id[-2:])/2))**2 /1000)  # 随机延迟1ms~[144,3844]ms，最大延迟由设备id决定\n",
    "    print(f\"{client_id}\\t接收: `{received_msg['msg']}`\")\n",
    "    delay_status[client_id][int(received_msg['msg'])] = (int((time.time() - received_msg['time']) * 1000))"
   ],
   "id": "89b5349380dda002",
   "outputs": [],
   "execution_count": 16
  },
  {
   "metadata": {},
   "cell_type": "markdown",
   "source": "#### 2.3.2 订阅节点",
   "id": "ce09b05b049b44cb"
  },
  {
   "metadata": {
    "ExecuteTime": {
     "end_time": "2025-04-16T02:50:20.503034Z",
     "start_time": "2025-04-16T02:50:20.490811Z"
    }
   },
   "cell_type": "code",
   "source": [
    "def subscriber_node():\n",
    "    def on_message(client, userdata, msg):  # 订阅端每次收到消息都会调用该（回调）函数\n",
    "        nonlocal t\n",
    "        t = 0\n",
    "\n",
    "        try:\n",
    "            recv = json.loads(msg.payload)\n",
    "        except json.JSONDecodeError:\n",
    "            print(f\"Error decoding message: {msg.payload}\")\n",
    "            return\n",
    "\n",
    "        if random.random() < 0.1:  # 客户端模拟10%丢包\n",
    "            print(f\"{client_id}\\t丢包: `{recv['msg']}`\")\n",
    "        else:\n",
    "            th.Thread(target=delay, args=(client_id, recv)).start()  # 另外开一个线程模拟延迟记录，避免延迟累加\n",
    "    global delay_status\n",
    "    t = 0 # 距离上一次接收消息过了多少秒\n",
    "\n",
    "    client_id = f'Subscriber-{random.randint(0, 999):0>3}' \t# 订阅设备的设备号\n",
    "\n",
    "    delay_status[client_id] = dict()\n",
    "    for i in range(batch_size):\n",
    "        delay_status[client_id][msg_id+i] = -1\n",
    "\n",
    "    client = connect_mqtt(client_id)\n",
    "    client.on_message = on_message\n",
    "    client.subscribe(topic)\n",
    "\n",
    "    client.loop_start()\n",
    "    # 等待订阅和消息接收\n",
    "    try:\n",
    "        timeout = 5\n",
    "        while t < timeout:\n",
    "            time.sleep(1)  # 该线程等待消息，创建的loop线程会接受消息\n",
    "            t += 1\n",
    "\n",
    "        print(f\"超时{timeout}s，关闭订阅端 {client_id}...\")\n",
    "    except KeyboardInterrupt:\n",
    "        print(f\"关闭订阅端 {client_id}...\")\n",
    "    finally:\n",
    "        client.loop_stop()  # 停止后台的MQTT网络循环\n",
    "        client.disconnect()"
   ],
   "id": "b321a32ae38cbc10",
   "outputs": [],
   "execution_count": 17
  },
  {
   "metadata": {},
   "cell_type": "markdown",
   "source": "### 3. 通过多线程创建节点",
   "id": "412ea875d2d7d4ee"
  },
  {
   "metadata": {
    "ExecuteTime": {
     "end_time": "2025-04-16T02:50:20.535430Z",
     "start_time": "2025-04-16T02:50:20.528884Z"
    }
   },
   "cell_type": "code",
   "source": [
    "# 创建发布/订阅者线程\n",
    "def create(target):\n",
    "    if target in (subscriber_node, publisher_node):\n",
    "        node_thread = th.Thread(target=target)\n",
    "        node_thread.daemon = True  # 设置为守护线程，主线程退出时，订阅者线程也会退出\n",
    "        node_thread.start()\n",
    "        return node_thread"
   ],
   "id": "ba6180f94cf357f3",
   "outputs": [],
   "execution_count": 18
  },
  {
   "metadata": {},
   "cell_type": "markdown",
   "source": "主要程序，可以循环多次运行",
   "id": "8e0957250e9e718d"
  },
  {
   "metadata": {
    "ExecuteTime": {
     "end_time": "2025-04-16T02:50:29.407077Z",
     "start_time": "2025-04-16T02:50:20.552595Z"
    }
   },
   "cell_type": "code",
   "source": [
    "nodes = [create(subscriber_node) for _ in range(3)]  # 创建多个订阅者\n",
    "time.sleep(1)  # 等待订阅端启动\n",
    "\n",
    "nodes.append(create(publisher_node))  # 创建发布者节点\n",
    "# 等待线程结束\n",
    "for node in nodes:\n",
    "    node.join()  # 主线程等待其他线程输出结束"
   ],
   "id": "65c3e5810d875fa5",
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Subscriber-692 connected to MQTT Broker!Subscriber-014 connected to MQTT Broker!\n",
      "\n",
      "Subscriber-356 connected to MQTT Broker!\n",
      "Publisher-27\t发送: `0` to topic `/python/mqtt/test`\n",
      "Publisher-27\t发送: `1` to topic `/python/mqtt/test`\n",
      "Publisher-27\t发送: `2` to topic `/python/mqtt/test`\n",
      "Publisher-27 connected to MQTT Broker!\n",
      "Publisher-27\t发送: `3` to topic `/python/mqtt/test`\n",
      "Subscriber-014\t接收: `1`\n",
      "Publisher-27\t发送: `4` to topic `/python/mqtt/test`\n",
      "Subscriber-356\t接收: `0`\n",
      "Publisher-27\t发送: `5` to topic `/python/mqtt/test`\n",
      "Publisher-27\t发送: `6` to topic `/python/mqtt/test`\n",
      "Subscriber-014\t接收: `2`\n",
      "Subscriber-692\t接收: `2`\n",
      "Publisher-27\t发送: `7` to topic `/python/mqtt/test`\n",
      "Subscriber-356\t丢包: `3`\n",
      "Subscriber-692\t接收: `3`\n",
      "Publisher-27\t发送: `8` to topic `/python/mqtt/test`\n",
      "Subscriber-014\t接收: `0`\n",
      "Subscriber-356\t丢包: `5`Publisher-27\t发送: `9` to topic `/python/mqtt/test`\n",
      "\n",
      "Subscriber-356\t接收: `4`\n",
      "Publisher-27\t发送: `10` to topic `/python/mqtt/test`\n",
      "Subscriber-356\t接收: `2`\n",
      "Subscriber-692\t接收: `0`\n",
      "Subscriber-014\t接收: `5`\n",
      "Publisher-27\t发送: `11` to topic `/python/mqtt/test`\n",
      "Subscriber-014\t接收: `3`\n",
      "Subscriber-692\t接收: `5`\n",
      "Subscriber-014\t接收: `7`\n",
      "Subscriber-692\t接收: `1`\n",
      "Publisher-27\t发送: `12` to topic `/python/mqtt/test`\n",
      "Publisher-27\t发送: `13` to topic `/python/mqtt/test`\n",
      "Subscriber-356\t接收: `9`\n",
      "Subscriber-014\t接收: `8`\n",
      "Publisher-27\t发送: `14` to topic `/python/mqtt/test`\n",
      "Publisher-27\t发送: `15` to topic `/python/mqtt/test`\n",
      "Subscriber-356\t接收: `8`\n",
      "Subscriber-014\t接收: `10`\n",
      "Subscriber-014\t接收: `11`\n",
      "Publisher-27\t发送: `16` to topic `/python/mqtt/test`\n",
      "Publisher-27\t发送: `17` to topic `/python/mqtt/test`\n",
      "Subscriber-014\t接收: `4`\n",
      "Subscriber-014\t丢包: `12`\n",
      "Publisher-27\t发送: `18` to topic `/python/mqtt/test`\n",
      "Subscriber-692\t接收: `7`\n",
      "Subscriber-014\t接收: `6`\n",
      "Publisher-27\t发送: `19` to topic `/python/mqtt/test`\n",
      "Subscriber-692\t接收: `12`\n",
      "Subscriber-014\t接收: `13`\n",
      "Subscriber-692\t丢包: `14`\n",
      "Subscriber-014\t接收: `9`\n",
      "Subscriber-356\t接收: `1`\n",
      "Subscriber-356\t接收: `15`\n",
      "Subscriber-692\t接收: `9`\n",
      "Subscriber-356\t丢包: `16`\n",
      "Subscriber-014\t丢包: `17`\n",
      "Subscriber-692\t接收: `16`\n",
      "Subscriber-014\t接收: `18`\n",
      "Subscriber-014\t接收: `16`\n",
      "Subscriber-692\t接收: `18`\n",
      "Subscriber-014\t接收: `19`\n",
      "Subscriber-356\t接收: `17`\n",
      "Subscriber-692\t接收: `13`\n",
      "Subscriber-356\t接收: `13`\n",
      "Subscriber-356\t接收: `10`\n",
      "Subscriber-692\t接收: `4`\n",
      "Subscriber-356\t接收: `7`\n",
      "Subscriber-356\t接收: `12`\n",
      "Subscriber-014\t接收: `14`\n",
      "Subscriber-014\t接收: `15`\n",
      "Subscriber-356\t接收: `6`\n",
      "Subscriber-692\t接收: `8`\n",
      "Subscriber-692\t接收: `19`\n",
      "Subscriber-692\t接收: `11`\n",
      "Subscriber-356\t接收: `14`\n",
      "Subscriber-356\t接收: `11`\n",
      "Subscriber-692\t接收: `15`\n",
      "Subscriber-356\t接收: `19`\n",
      "Subscriber-356\t接收: `18`\n",
      "Subscriber-692\t接收: `17`\n",
      "Subscriber-692\t接收: `6`\n",
      "Subscriber-692\t接收: `10`\n",
      "超时5s，关闭订阅端 Subscriber-692...\n",
      "超时5s，关闭订阅端 Subscriber-014...\n",
      "超时5s，关闭订阅端 Subscriber-356...\n"
     ]
    }
   ],
   "execution_count": 19
  },
  {
   "metadata": {},
   "cell_type": "markdown",
   "source": "### 4. 统计结果",
   "id": "22706f384053800"
  },
  {
   "metadata": {
    "ExecuteTime": {
     "end_time": "2025-04-16T02:50:29.423892Z",
     "start_time": "2025-04-16T02:50:29.415474Z"
    }
   },
   "cell_type": "code",
   "source": [
    "total_count = 0\n",
    "total_delay_sum = 0\n",
    "delayed_count = 0\n",
    "for client_id, ds in delay_status.items():\n",
    "    delay_status[client_id] = dict(sorted(ds.items(), key=lambda x: x[0]))\n",
    "    print(f'{client_id}: {delay_status[client_id]}')\n",
    "\n",
    "    private_total_count = len(ds)\n",
    "    total_count += private_total_count\n",
    "    # 统计延迟\n",
    "    delay_sum = 0\n",
    "    private_delayed_count = 0\n",
    "    for k,v in ds.items():\n",
    "        if v >= 0:\n",
    "            delay_sum += v\n",
    "            private_delayed_count += 1\n",
    "\n",
    "    total_delay_sum += delay_sum\n",
    "    delayed_count += private_delayed_count\n",
    "    print(f'总共向 {client_id} 发送{private_total_count}条消息\\t'\n",
    "          f'平均延迟{delay_sum/private_delayed_count :.2f}\\t'\n",
    "          f'丢包：{private_total_count-private_delayed_count}次，丢包率{(private_total_count-private_delayed_count)*100/private_total_count :.2f}%\\n')\n",
    "\n",
    "# 在忽略计算延迟的情况下，设备id为x75时理论平均延迟约为844ms，然而实际延迟还要加上与Broker代理服务器的通信延迟，平均延迟会增加300~800ms，具体视本机情况而定\n",
    "print(\n",
    "    f'总发送：{total_count}\\t',\n",
    "    f'平均延迟：{total_delay_sum / delayed_count :.2f}ms\\t',\n",
    "    f'丢包：{total_count-delayed_count}次，丢包率{(total_count-delayed_count)*100/total_count :.2f}%',\n",
    ")"
   ],
   "id": "fa6edbdd72f0ed4",
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Subscriber-692: {0: 1098, 1: 1108, 2: 497, 3: 436, 4: 2272, 5: 664, 6: 4915, 7: 1135, 8: 2394, 9: 1356, 10: 5045, 11: 2392, 12: 727, 13: 1211, 14: -1, 15: 2564, 16: 754, 17: 3678, 18: 614, 19: 1585}\n",
      "总共向 Subscriber-692 发送20条消息\t平均延迟1812.89\t丢包：1次，丢包率5.00%\n",
      "\n",
      "Subscriber-356: {0: 450, 1: 1949, 2: 811, 3: -1, 4: 594, 5: -1, 6: 2494, 7: 1980, 8: 747, 9: 473, 10: 1579, 11: 2663, 12: 1607, 13: 1212, 14: 2303, 15: 691, 16: -1, 17: 793, 18: 2951, 19: 2571}\n",
      "总共向 Subscriber-356 发送20条消息\t平均延迟1521.65\t丢包：3次，丢包率15.00%\n",
      "\n",
      "Subscriber-014: {0: 809, 1: 284, 2: 409, 3: 827, 4: 1350, 5: 603, 6: 1287, 7: 467, 8: 602, 9: 1131, 10: 556, 11: 464, 12: -1, 13: 679, 14: 1463, 15: 1485, 16: 816, 17: -1, 18: 566, 19: 561}\n",
      "总共向 Subscriber-014 发送20条消息\t平均延迟797.72\t丢包：2次，丢包率10.00%\n",
      "\n",
      "总发送：60\t 平均延迟：1382.81ms\t 丢包：6次，丢包率10.00%\n"
     ]
    }
   ],
   "execution_count": 20
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
